﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

 
	public class AsynQueue<T>
	{
		//队列是否正在处理数据
		private int isProcessing;
		//有线程正在处理数据
		private const int Processing = 1;
		//没有线程处理数据
		private const int UnProcessing = 0;
		//队列是否可用
		private volatile bool enabled = true;
		private Task currentTask;
		public event Action<T> ProcessItemFunction;
		public event EventHandler<EventArgs<Exception>> ProcessException;
		private ConcurrentQueue<T> queue;

		public AsynQueue()
		{
			queue = new ConcurrentQueue<T>();
			Start();
		}

		public int Count
		{
			get
			{
				return queue.Count;
			}
		}

		private void Start()
		{
			Thread process_Thread = new Thread(PorcessItem);
			process_Thread.IsBackground = true;
			process_Thread.Start();
		}

		public void Enqueue(T items)
		{
			if (items == null)
			{
				throw new ArgumentException("items");
			}

			queue.Enqueue(items);
			DataAdded();
		}

		//数据添加完成后通知消费者线程处理
		private void DataAdded()
		{
			if (enabled)
			{
				if (!IsProcessingItem())
				{
					currentTask = Task.Factory.StartNew(ProcessItemLoop);
				}
			}
		}

		//判断是否队列有线程正在处理 
		private bool IsProcessingItem()
		{
			return !(Interlocked.CompareExchange(ref isProcessing, Processing, UnProcessing) == 0);
		}

		private void ProcessItemLoop()
		{

			if (!enabled && queue.IsEmpty)
			{
				Interlocked.Exchange(ref isProcessing, 0);
				return;
			}
			int runingCore = 0;
			var count = Thread.VolatileRead(ref runingCore);
			//处理的线程数 是否小于当前最大任务数
			//if (Thread.VolatileRead(ref runingCore) <= this.MaxTaskCount)
			//{
			T publishFrame;

			if (queue.TryDequeue(out publishFrame))
			{

				try
				{
					ProcessItemFunction(publishFrame);

				}
				catch (Exception ex)
				{
					OnProcessException(ex);
				}
			}

			if (enabled && !queue.IsEmpty)
			{
				currentTask = Task.Factory.StartNew(ProcessItemLoop);
			}
			else
			{
				Interlocked.Exchange(ref isProcessing, UnProcessing);
			}
		}

		/// <summary>
		///定时处理线程调用函数  
		///主要是监视入队的时候线程 没有来的及处理的情况
		/// </summary>
		private void PorcessItem(object state)
		{
			int sleepCount = 0;
			int sleepTime = 1000;
			while (enabled)
			{
				//如果队列为空则根据循环的次数确定睡眠的时间
				if (queue.IsEmpty)
				{
					if (sleepCount == 0)
					{
						sleepTime = 1000;
					}
					else if (sleepCount <= 3)
					{
						sleepTime = 1000 * 3;
					}
					else
					{
						sleepTime = 1000 * 50;
					}
					sleepCount++;
					Thread.Sleep(sleepTime);
				}
				else
				{
					//判断是否队列有线程正在处理 
					if (enabled && Interlocked.CompareExchange(ref isProcessing, Processing, UnProcessing) == 0)
					{
						if (!queue.IsEmpty)
						{
							currentTask = Task.Factory.StartNew(ProcessItemLoop);
						}
						else
						{
							Interlocked.Exchange(ref isProcessing, 0);
						}
						sleepCount = 0;
						sleepTime = 1000;
					}
				}
			}
		}

		public void Flsuh()
		{
			Stop();

			if (currentTask != null)
			{
				currentTask.Wait();
			}

			while (!queue.IsEmpty)
			{
				try
				{
					T publishFrame;
					if (queue.TryDequeue(out publishFrame))
					{
						ProcessItemFunction(publishFrame);
					}
				}
				catch (Exception ex)
				{
					OnProcessException(ex);
				}
			}
			currentTask = null;
		}

		public void Stop()
		{
			this.enabled = false;
		}

		private void OnProcessException(System.Exception ex)
		{
			var tempException = ProcessException;
			Interlocked.CompareExchange(ref ProcessException, null, null);

			if (tempException != null)
			{
				ProcessException(ex, new EventArgs<Exception>(ex));
			}
		}

	[Serializable]
	public class EventArgs<T> : System.EventArgs
	{
		public T Argument;

		public EventArgs() : this(default(T))
		{
		}

		public EventArgs(T argument)
		{
			Argument = argument;
		}
	}
}

	public class ProcessQueue<T>
	{
		private BlockingCollection<T> _queue;
		private CancellationTokenSource _cancellationTokenSource;
		private CancellationToken _cancellToken;
		//内部线程池
		private List<Thread> _threadCollection;
		private readonly int _millisecondsTimeout;//线程处理超时时间(毫秒)

		//队列是否正在处理数据
		private int _isProcessing;
		//有线程正在处理数据
		private const int Processing = 1;
		//没有线程处理数据
		private const int UnProcessing = 0;
		//队列是否可用   单个线程下用while来判断，多个线程下用if判断，随后while循环队列的数量
		private volatile bool _enabled = true;
		//内部处理线程数量
		private int _internalThreadCount;
		// 消费者处理事件
		public event Action<T> ProcessItemEvent;
		//处理异常，需要三个参数，当前队列实例，异常，当时处理的数据
		public event Action<dynamic, Exception, T> ProcessExceptionEvent;
		public event Action<dynamic, Exception, T> ProcessTimeOutExceptionEvent;


		public ProcessQueue() : this(10,10*60*1000)
		{ 
		}

		public ProcessQueue(int internalThreadCount,int millisecondsTimeout) 
		{
			this._millisecondsTimeout = millisecondsTimeout;
			this._internalThreadCount = internalThreadCount;

			_queue = new BlockingCollection<T>();
			_cancellationTokenSource = new CancellationTokenSource();
			_internalThreadCount = 10;
			_cancellToken = _cancellationTokenSource.Token;
			_threadCollection = new List<Thread>();
		}

		/// <summary>
		/// 队列内部元素的数量 
		/// </summary>
		public int GetInternalItemCount()
		{
			//return _queue.Count;
			return _threadCollection.Count;
		}
		public int GetQueueCount()
		{
			return _queue.Count;

		}
		//生产者生产
		public void Enqueue(T items)
		{
			if (items == null)
			{
				throw new ArgumentException("items");
			}

			_queue.Add(items);
			DataAdded();
		}

		public void Flush()
		{
			StopProcess();

			while (_queue.Count != 0)
			{
				T item = default(T);
				if (_queue.TryTake(out item))
				{
					try
					{
						ProcessItemEvent(item);
					}
					catch (Exception ex)
					{
						OnProcessException(ex, item);
					}
				}
			}
		}
		// 通知消费者消费队列元素
		private void DataAdded()
		{
			if (_enabled)
			{
				if (!IsProcessingItem())
				{
					Console.WriteLine("DataAdded");
					ProcessRangeItem();
					StartProcess();
				}
			}
		}

		//判断是否队列有线程正在处理 
		private bool IsProcessingItem()
		{
			// 替换第一个参数, 如果相等
			//int x = Interlocked.CompareExchange(ref _isProcessing, Processing, UnProcessing);
			return !(Interlocked.CompareExchange(ref _isProcessing, Processing, UnProcessing) == UnProcessing);
		}
		// 多消费者消费
		private void ProcessRangeItem()
		{
			for (int i = 0; i < this._internalThreadCount; i++)
			{
				ProcessItem();
			}
		}
		// 开启消费处理
		private void ProcessItem()
		{
			Thread currentThread = new Thread((state) =>
			{
				T item = default(T);
				while (_enabled)
				{
					try
					{
						if (!_queue.TryTake(out item))
						{
							//Console.WriteLine("阻塞队列为0时的item: {0}", item);
							//Console.WriteLine("ok!!!");
							continue;
							break;
						}
						// 处理事件
						TimeoutChecker timeout = new TimeoutChecker(
					delegate
					{
						ProcessItemEvent(item);

					},
					delegate//超时处理
					{
						OnProcessTimeOutException(
								new Exception("执行超时"), item);
					});

						// 按毫秒等待
						// 执行成功的处理,且未超时
						if (timeout.Wait(_millisecondsTimeout))
						{
							
						}


					}
					catch (Exception ex)
					{
						OnProcessException(ex, item);
					}
				}
			});
			_threadCollection.Add(currentThread);
		}
		// 开启消费者
		private void StartProcess()
		{
			//Console.WriteLine("线程的数量: {0}", _threadCollection.Count);
			foreach (var thread in _threadCollection)
			{
				thread.Start();
				thread.IsBackground = true;
			}
		}
		// 终止运行
		private void StopProcess()
		{
			this._enabled = false;
			foreach (var thread in _threadCollection)
			{
				if (thread.IsAlive)
				{
					thread.Join();
				}
			}
			_threadCollection.Clear();
		}
		private void OnProcessTimeOutException(Exception ex, T item)
		{
			var tempException = ProcessTimeOutExceptionEvent;
			Interlocked.CompareExchange(ref ProcessTimeOutExceptionEvent, null, null);

			if (tempException != null)
			{
				ProcessTimeOutExceptionEvent(this, ex, item);
			}
		}
		private void OnProcessException(Exception ex, T item)
		{
			var tempException = ProcessExceptionEvent;
			Interlocked.CompareExchange(ref ProcessExceptionEvent, null, null);

			if (tempException != null)
			{
				ProcessExceptionEvent(this, ex, item);
			}
		}

	}

	public class TimeoutChecker
	{
		#region /*private member*/

		long _timeout; //超时时间
		System.Action<Delegate> _proc;//会超时的代码
		System.Action<Delegate> _procHandle;//处理超时
		System.Action<Delegate> _timeoutHandle;//超时后处理事件
		System.Threading.ManualResetEvent _event = new System.Threading.ManualResetEvent(false);

		#endregion

		#region /*constructor method*/

		/// <summary>
		/// 构选方法
		/// </summary>
		/// <param name="proc">会超时的代码</param>
		/// <param name="timeoutHandle">超时后处理事件</param>
		public TimeoutChecker(System.Action<Delegate> proc, System.Action<Delegate> timeoutHandle)
		{
			this._proc = proc;
			this._timeoutHandle = timeoutHandle;
			this._procHandle = delegate
			{
				//计算代码执行的时间
				System.Diagnostics.Stopwatch sw = new System.Diagnostics.Stopwatch();
				sw.Start();
				if (this._proc != null)
					this._proc(null);
				sw.Stop();
				//如果执行时间小于超时时间则通知用户线程
				if (sw.ElapsedMilliseconds < this._timeout && this._event != null)
				{
					this._event.Set();
				}
			};
		}

		#endregion

		#region /*public method*/

		/// <summary>
		/// 等待执行
		/// </summary>
		/// <param name="timeout">等待时间毫秒</param>
		/// <returns></returns>
		public bool Wait(long timeout)
		{
			this._timeout = timeout;
			//异步执行
			this._procHandle.BeginInvoke(null, null, null);
			//如果在规定时间内没等到通知则为 false
			bool flag = this._event.WaitOne((int)timeout, false);
			if (!flag)
			{
				//触发超时时间
				if (this._timeoutHandle != null)
					this._timeoutHandle(null);
			}
			this.Dispose();

			return flag;
		}

		#endregion

		#region private method

		/// <summary>
		/// 释放资源
		/// </summary>
		private void Dispose()
		{
			if (this._event != null)
				this._event.Close();
			this._event = null;
			this._proc = null;
			this._procHandle = null;
			this._timeoutHandle = null;
		}

		#endregion
	}
 
